package com.ssy.jt808.gateway.infrastructure.gateway.impl;

import com.ssy.gaea.message.GaeaMessage;
import com.ssy.gaea.message.GaeaMessageBody;
import com.ssy.gaea.util.JsonMapper;
import com.ssy.jt808.gateway.infrastructure.gateway.TerminalMessagePublisher;
import com.ssy.jt808.gateway.sdk.terminal.constant.AlarmMessageEnum;
import com.ssy.jt808.gateway.sdk.terminal.constant.ConnectionStatusEnum;
import com.ssy.jt808.gateway.sdk.terminal.constant.LocationMessageEnum;
import com.ssy.jt808.gateway.sdk.terminal.constant.MessageTopicConstant;
import com.ssy.jt808.gateway.sdk.terminal.message.location.T0200DataMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * @author wangzezhou
 */
@Component
public class TerminalMessagePublisherKafkaImpl implements TerminalMessagePublisher{
  private KafkaTemplate<Long,String> kafkaTemplate;

  @Override
  public void sendConnectionStatus( GaeaMessageBody<ConnectionStatusEnum> messageBody, Long routingKey ){
    GaeaMessage<ConnectionStatusEnum,GaeaMessageBody<ConnectionStatusEnum>> connectionStatusMessage = new GaeaMessage<>(
    messageBody.type(),
    messageBody);
    kafkaTemplate.send(MessageTopicConstant.CONNECTION_STATUS,
                       routingKey,
                       JsonMapper.toJsonString(connectionStatusMessage));
  }

  @Override
  public void sendLocations( GaeaMessageBody<LocationMessageEnum> messageBody, Long routingKey ){

    GaeaMessage<LocationMessageEnum,GaeaMessageBody<LocationMessageEnum>> locationMessage = new GaeaMessage<>(
    messageBody.type(),
    messageBody);
    kafkaTemplate.send(MessageTopicConstant.LOCATION_INFO, routingKey, JsonMapper.toJsonString(locationMessage));
  }

  @Override
  public void sendAlarmInfo( GaeaMessageBody<AlarmMessageEnum> messageBody, Long routingKey ){

    GaeaMessage<AlarmMessageEnum,GaeaMessageBody<AlarmMessageEnum>> alarmMessage = new GaeaMessage<>(messageBody.type(),
                                                                                                     messageBody);
    kafkaTemplate.send(MessageTopicConstant.ALARM_INFO, routingKey, JsonMapper.toJsonString(alarmMessage));
  }

  /**
   * 兼容性消息推送
   *
   * @param message
   */
  @Override
  public void sendT0200Messsage( T0200DataMessage message, Long routingKey ){
    kafkaTemplate.send(MessageTopicConstant.T0200_DATA,
                       routingKey,
                       JsonMapper.toJsonString(message));
  }

  @Autowired
  public void setKafkaTemplate( KafkaTemplate<Long,String> kafkaTemplate ){
    this.kafkaTemplate = kafkaTemplate;
  }

}
